package Autodesk;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Hashtable;
import java.util.List;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class TotalRequestVolume extends Configured implements Tool  {
	
		public Path prodPath;
		public static class LogMapper extends Mapper<Object, Text, Text, IntWritable> {
			
			private final static IntWritable one = new IntWritable(1);
			private Text word = new Text();
			private String interval;
			private String param1;
			private String param2="";
			Hashtable<String, String> ht = new Hashtable<String, String>();
			
			public void setup(Context context) {
				
				Configuration conf = context.getConfiguration();
				String paramNotReadyInt = conf.get("input.interval");
				interval = paramNotReadyInt.toString().toLowerCase();
				String paramNotReady = conf.get("input.parameter1");
				param1 = paramNotReady.toString().toLowerCase();
				if (conf.getBoolean("input2", false)) {
					String paramNotReady2 = conf.get("input.parameter2");
					param2 = paramNotReady2.toString().toLowerCase();
				}
			
				
			}

			public void map(Object key, Text value, Context context)
					throws IOException, InterruptedException {
				
				String record[] = value.toString().toLowerCase().split("\t", 11);
				StringBuffer bf = new StringBuffer();
				if(("daily".equals(interval)) || ("1".equals(interval))){
					bf.append(record[0]);//This get the date.
					bf.append("\t");
				}
				if("".equals(param2)){
					if("language".equals(param1) || "languages".equals(param1)){
						ht.put("l", record[7]);
					}else if("channel".equals(param1) || "channels".equals(param1)){
						ht.put("c", record[1]);
					}else if("product".equals(param1) || "products".equals(param1)){
						ht.put("p", record[5]);					}
				}else{
					if( ("language".equals(param1) || "languages".equals(param1)) & ("product".equals(param2) || "products".equals(param2)) || 
							("language".equals(param2) || "languages".equals(param2)) & ("product".equals(param1) || "products".equals(param1)) ){
						ht.put("l", record[7]);
						ht.put("p", record[5]);
					}else if(("channel".equals(param1) || "channels".equals(param1)) & ("product".equals(param2) || "products".equals(param2)) || 
							("channel".equals(param2) || "channels".equals(param2)) & ("product".equals(param1) || "products".equals(param1))){
						ht.put("c", record[1]);
						ht.put("p", record[5]);
					}else if(("channel".equals(param1) || "channels".equals(param1)) & ("language".equals(param2) || "languages".equals(param2)) || 
							("channel".equals(param2) || "channels".equals(param2)) & ("language".equals(param1) || "languages".equals(param1)) ){
						ht.put("c", record[1]);
						ht.put("l", record[7]);
					}
				}
				if (ht != null && !ht.isEmpty()) {
					bf.append((ht.get("c") != null && !ht.get("c").isEmpty()) ? ht.get("c").toString() : "\t");
					bf.append("\t");
					bf.append((ht.get("l") != null && !ht.get("l").isEmpty()) ? ht.get("l").toString() : "\t");
					bf.append("\t");
					bf.append((ht.get("p") != null && !ht.get("p").isEmpty()) ? ht.get("p").toString() : "\t");
					bf.append("\t");
					word.set(bf.toString());
					context.write(word, one);
				}
			}

	}


	public static class LogReducer extends
			Reducer<Text, IntWritable, Text, IntWritable> {
		
		private IntWritable result = new IntWritable();

		public void reduce(Text key, Iterable<IntWritable> values,
				Context context) throws IOException, InterruptedException {
			
			int sum = 0;
			for (IntWritable val : values) {
				sum += val.get();
			}
			result.set(sum);
			context.write(key, result);
		
		}
	}
	
	public int run(String[] args) throws Exception { 
		Configuration conf = getConf();
		
		Job job = new Job(conf, "ParseLogs");
		job.setJarByClass(Autodesk.TotalRequestVolume.class);
		
//		FileInputFormat.addInputPath(job, new Path(args[0]));
//		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		
		job.setMapperClass(LogMapper.class);
		job.setReducerClass(LogReducer.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		List<String> other_args = new ArrayList<String>();
		for (int i = 0; i < args.length; ++i) {
			if ("-Interval".equals(args[i])) {
				job.getConfiguration().set("input.interval", new String(args[++i]));
			}else if ("-Param1".equals(args[i])) {
				job.getConfiguration().set("input.parameter1", new String(args[++i]));
			}else if ("-Param2".equals(args[i])) {
				job.getConfiguration().setBoolean("input2",true);
				job.getConfiguration().set("input.parameter2", new String(args[++i]));
			}else{
				other_args.add(args[i]);
			}
		}
	    FileInputFormat.setInputPaths(job, new Path(other_args.get(0)));
	    FileOutputFormat.setOutputPath(job, new Path(other_args.get(1)));

		System.exit(job.waitForCompletion(true)?0:1);
		
		return 0;
	
	}
	
	public static void main(String[] args) throws Exception {
		int res = ToolRunner.run(new Configuration(), new TotalRequestVolume(), args);
		System.exit(res);
		}

 
}
